package com.ls.io.reactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class SelectorThread implements Runnable{

    Selector selector;

    SelectorThreadGroup group;

    LinkedBlockingQueue<Channel> channelQueue = new LinkedBlockingQueue<>();

    public SelectorThread(SelectorThreadGroup group){
        this.group = group;
        try {
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void run() {

        while (true){
            try {
                int num = selector.select(5000);
//                System.out.println("select..." + num);

                if(num > 0){
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();

                    while (iterator.hasNext()){
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();

                        if(selectionKey.isAcceptable()){
                            acceptHandler(selectionKey);
                        }else if(selectionKey.isReadable()){
                            readHandler(selectionKey);
                        }else if(selectionKey.isWritable()){
                            writeHandler(selectionKey);
                        }
                    }
                }

                if(!channelQueue.isEmpty()){
                    Channel channel = null;
                    while((channel = channelQueue.poll()) != null){
                        if(channel instanceof ServerSocketChannel){
                            System.out.println(Thread.currentThread().getName() + ":  listen");
                            ((ServerSocketChannel) channel)
                                    .register(selector, SelectionKey.OP_ACCEPT);
                        }else{
                            ((SocketChannel)channel)
                                    .register(selector, SelectionKey.OP_READ, ByteBuffer.allocateDirect(1024));
                        }
                    }
                }

            } catch (IOException e) {
                e.printStackTrace();
            }

        } // end for while

    }

    private void acceptHandler(SelectionKey selectionKey) {
        System.out.println(Thread.currentThread().getName() + ":  register");
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);

            group.nextSelector(socketChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void readHandler(SelectionKey selectionKey) throws IOException {
        System.out.println(Thread.currentThread().getName() + ":  read");
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
        buffer.clear();

        int num = 0;
        while((num = socketChannel.read(buffer)) > 0){
            buffer.flip();
            socketChannel.write(buffer);

            buffer.clear();
        }

    }

    private void writeHandler(SelectionKey selectionKey) {

    }
}
